package main

import (
	"flag"
	"log"
	"os"
	"runtime"
	"runtime/pprof"
	"time"

	console "github.com/asynkron/goconsole"
	"github.com/asynkron/protoactor-go/actor"
	"github.com/asynkron/protoactor-go/remote"

	"learn-go/protoactor-go/remote-benchmark-2/messages"
)

const pingTimes = 1000000

type localActor struct {
	remote    *remote.Remote
	remotePid *actor.PID
	count     int32
	beginTime time.Time
}

func newLocalActor(remote *remote.Remote) actor.Producer {
	return func() actor.Actor {
		return &localActor{
			remote: remote,
		}
	}
}

func (a *localActor) ping(context actor.Context) {
	context.Send(a.remotePid, &messages.Ping{Count: a.count})
}

func (a *localActor) Receive(context actor.Context) {
	switch msg := context.Message().(type) {
	case *actor.Started:
		pidResponse, err := a.remote.Spawn("127.0.0.1:12000", "echo", 2000*time.Second)
		if err != nil || pidResponse.StatusCode != 0 {
			context.Stop(context.Self())
			return
		}
		a.remotePid = pidResponse.Pid

		_ = context.RequestFuture(a.remotePid, &messages.StartRemote{Sender: context.Self()}, 5*time.Second).Wait()

		a.count = 1
		a.beginTime = time.Now()

		a.ping(context)

	case *messages.Pong:
		a.count = msg.Count + 1
		if a.count%10000 == 0 {
			log.Printf("%d use time: %v", a.count, time.Since(a.beginTime))
		}
		if a.count >= pingTimes {
			useTime := time.Since(a.beginTime)
			qps := int(float32(pingTimes) / (float32(useTime) / float32(time.Second)))
			log.Printf("%d done qps: %v", a.count, qps)
			a.count = 1
			a.beginTime = time.Now()
		}
		a.ping(context)
	}
}

var (
	cpuProfile   = flag.String("cpuprofile", "", "write cpu profile to file")
	blockProfile = flag.String("blockprof", "", "execute contention profiling and save results here")
)

func main() {
	flag.Parse()
	if *cpuProfile != "" {
		f, err := os.Create(*cpuProfile)
		if err != nil {
			log.Fatal(err)
		}
		_ = pprof.StartCPUProfile(f)
		defer pprof.StopCPUProfile()
	}

	// Check for lock contention profiling
	if *blockProfile != "" {
		prof, err := os.Create(*blockProfile)
		if err != nil {
			log.Fatal(err)
		}
		runtime.SetBlockProfileRate(1)
		defer func() {
			_ = pprof.Lookup("block").WriteTo(prof, 0)
		}()
	}

	// runtime.GOMAXPROCS(runtime.NumCPU() * 1)
	// runtime.GC()

	// remote.DefaultSerializerID = 1
	system := actor.NewActorSystem()
	r := remote.NewRemote(system, remote.Configure("127.0.0.1", 8081 /*, remote.WithCallOptions(grpc.UseCompressor(gzip.Name))*/))
	r.Start()

	rootContext := system.Root

	props := actor.
		PropsFromProducer(newLocalActor(r)) /*,
		actor.WithMailbox(actor.Bounded(1000000))*/

	pid := rootContext.Spawn(props)

	_, _ = console.ReadLine()

	rootContext.Stop(pid)

	r.Shutdown(true)
}
